package com.shujia.stream

import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo1StreamWordCount {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("stream")
      .setMaster("local[2]") //指定两个线程

    val sc = new SparkContext(conf)

    //创建spark streaming上下文对象   指定间隔时间
    val ssc = new StreamingContext(sc, Durations.seconds(5))

    /**
      * checkpoint  将rdd数据写入到hdfs
      * cache 将数据写入Executor
      *
      */

    //设置状态保存地址
    ssc.checkpoint("data/checkpoint")

    //读取实时数据创建DStream
    //在linux中执行  nc -lk 8888   开启socket服务
    val ds = ssc.socketTextStream("192.168.129.101", 8888)

    /*val countDS = ds
      .flatMap(_.split(","))
      .map((_, 1))
      .reduceByKey(_ + _) //统计当前batch结果*/

    /**
      * 有状态计算
      */

    /**
      * seq  当前batch 一个key value的 所有数据
      * opt  之前batch计算结果（状态）   状态存储在hdfs
      *
      */


    val updataFun = (seq: Seq[Int], opt: Option[Int]) => {
      //当前batch统计结果
      val currCount = seq.sum

      //之前所有batch的状态   如果没有状态  返回一个默认值
      val last = opt.getOrElse(0)


      //将当前batch的结果加上之前所有batch的结果，得到总的结果
      val sum = currCount + last

      //返回一个更新之后的状态
      Option(sum)
    }

    val countDS = ds
      .flatMap(_.split(","))
      .map((_, 1))
      .updateStateByKey(updataFun) //有状态计算

    //打印实时数据
    countDS.print()


    //启动spark streaming程序
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()


  }
}
